package com.rainsoft.center.isec.frame.record;

import com.alibaba.fastjson.JSONObject;
import com.rainsoft.center.isec.frame.entity.AppConfig;
import com.rainsoft.center.isec.frame.start.PrepareSpark;
import com.rainsoft.center.isec.stream.content.entity.AHBaseEntity;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Map;
import java.util.Objects;

/**
 * @Package com.rainsoft.center.isec.frame
 * @Description:
 * @Author elwyn
 * @Date 2017/12/9 22:36
 * @Email elonyong@163.com
 */
public abstract class ARecordDataJob<T extends AHBaseEntity> implements Serializable {
    protected Class<T> clazz;
    protected PrepareSpark prepareSpark;
    private JavaInputDStream<ConsumerRecord<String, String>> javaInputDStream;
    protected AppConfig appConfig;

    public ARecordDataJob(Class<T> clazz, PrepareSpark prepareSpark) {
        this.appConfig = prepareSpark.getAppConfig();
        this.clazz = clazz;
        this.prepareSpark = prepareSpark;
        this.javaInputDStream = initJavaInputDStream();
    }

    private JavaInputDStream<ConsumerRecord<String, String>> initJavaInputDStream() {
        JavaStreamingContext javaStreamingContext = prepareSpark.getJavaStreamingContext();
        Map<String, Object> kafkaParams = prepareSpark.getKafkaParams();
        return KafkaUtils.createDirectStream(javaStreamingContext, LocationStrategies.PreferBrokers(),
                ConsumerStrategies.Subscribe(Arrays.asList(appConfig.getTopic()), kafkaParams));
    }


    protected JavaDStream<T> getJavaDStream() {
        return javaInputDStream.map(v1 -> {
            try {
                return JSONObject.parseObject(v1.value(), clazz);
            } catch (Exception e) {
                e.printStackTrace();
                return null;
            }
        }).filter(Objects::nonNull);
    }


    public abstract void foreachDStream();
}
